package org.teemor.demo.kafka.javaclient.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * @Description:
 * @author zhoulk
 * @date: 2020/2/14
 */
public class SimpleConsumer {
    public boolean isTrue = true;

    public static void main(String[] args) {
        SimpleConsumer simpleProducer = new SimpleConsumer();
        simpleProducer.simpleConsume();
    }

    public void simpleConsume() {
        Properties props = new Properties();
        props.put("bootstrap.servers", "common.teemor.xyz:9091");
        props.put("group.id", "test1111");
        props.put("enable.auto.commit", "true");
        props.put("connections.max.idle.ms", 10000);
        props.put("auto.commit.interval.ms", "1000");
        props.put("max.poll.records", "100");
        props.put("fetch.min.bytes", "1000");
        props.put(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG, "org.teemor.demo.kafka.javaclient.consumer.SimpleInterceptor,org.teemor.demo.kafka.javaclient.consumer.SimpleInterceptor2");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        Map<TopicPartition, OffsetAndMetadata> currentOfssets = new HashMap<>();
        consumer.subscribe(Arrays.asList("topic-footprint"), new ConsumerRebalanceListener() {
            @Override
            public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
                //store soffset in DB (currentOfssets)
            }

            @Override
            public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
                for (TopicPartition partition : partitions) {
//                    int offset = getOffsetFromDB(partition) ;
//                    consumer.seek(partition, offset);
                }

            }
        });
        TopicPartition  partition = new TopicPartition("topic-footprint", 0);
        ConsumerRecords<String, String> records = consumer.poll(Duration.ofSeconds(1));

        while (isTrue) {
            records = consumer.poll(Duration.ofSeconds(10));
            for (ConsumerRecord<String, String> record : records) {
                currentOfssets.put(new TopicPartition(record.topic(),record.partition()),new OffsetAndMetadata(record.offset()));
                System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
            }
            consumer.commitAsync();
        }
    }
}
